#pragma once

#include <iostream>
#include <functional>
#include <memory>
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpServer.h"
#include "mq_connection.hpp"
#include "mq_consumer.hpp"
#include "mq_virtualhost.hpp"
#include "../mqcommon/mq_threadpool.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include "../mqcommon/mq_proto.pb.h"
#include "../mqcommon/mq_logger.hpp"

namespace bitmq
{
    #define DBFILE "/meta.db"
    class MQBrokerServer
    {
    public:
        typedef std::shared_ptr<google::protobuf::Message> MessagePtr;
        MQBrokerServer(int port, const std::string &basedir)
            : _server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port), 
                "Server", muduo::net::TcpServer::kReusePort)
            , _dispatcher(std::bind(&MQBrokerServer::onUnknownMessage, this, std::placeholders::_1, 
                std::placeholders::_2, std::placeholders::_3))
            , _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)))
            , _virtual_host(std::make_shared<VirtualHost>("Virtual_Host", basedir, basedir + DBFILE))
            , _consumer_manager(std::make_shared<ConsumerManager>())
            , _connection_manager(std::make_shared<ConnectionManager>())
            , _threadpool(std::make_shared<threadpool>())
        {
            //针对历史消息中的所有队列，初始化队列的消费者管理结构
            auto qm = _virtual_host->allQueues();
            for (auto& q : qm)
            {
                _consumer_manager->initQueueConsumer(q.first);
            }

            //注册业务处理函数
            _dispatcher.registerMessageCallback<openChannelRequest>(std::bind(&MQBrokerServer::onOpenChannel, this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<closeChannelRequest>(std::bind(&MQBrokerServer::onCloseChannel, this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<declareExchangeRequest>(std::bind(&MQBrokerServer::onDeclareExchange, this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<deleteExchangeRequest>(std::bind(&MQBrokerServer::onDeleteExchange, this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<declareQueueRequest>(std::bind(&MQBrokerServer::onDeclareQueue, this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<deleteQueueRequest>(std::bind(&MQBrokerServer::onDeleteQueue, this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<queueBindRequest>(std::bind(&MQBrokerServer::onQueueBind, this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<queueUnBindRequest>(std::bind(&MQBrokerServer::onQueueUnBind, this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<basicPublishRequest>(std::bind(&MQBrokerServer::onBasicPublish, this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)); 
            _dispatcher.registerMessageCallback<basicAckRequest>(std::bind(&MQBrokerServer::onBasicAck, this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));  
            _dispatcher.registerMessageCallback<basicConsumeRequest>(std::bind(&MQBrokerServer::onBasicConsumer, this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));  
            _dispatcher.registerMessageCallback<basicCancelRequest>(std::bind(&MQBrokerServer::onBasicCancel, this,
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));             
            //设置回调函数
            _server.setConnectionCallback(std::bind(&MQBrokerServer::onConnection, this, std::placeholders::_1));
            _server.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(), std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        }

        void start()
        {
            _server.start();
            _baseloop.loop();
        }
    
    private:
        //打开信道
        void onOpenChannel(const muduo::net::TcpConnectionPtr& conn, const openChannelRequestPtr& req, muduo::Timestamp)
        {
            //先获取Connection对象
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr)
            {
                DLOG("打开信道失败，没有找到对应的Connection对象");
                conn->shutdown();
                return;
            }
            return mconn->openChannel(req);
        }

        //关闭信道
        void onCloseChannel(const muduo::net::TcpConnectionPtr& conn, const closeChannelRequestPtr& req, muduo::Timestamp)
        {
            //先获取Connection对象
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr)
            {
                DLOG("关闭信道失败，没有找到对应的Connection对象");
                conn->shutdown();
                return;
            }
            return mconn->closeChannel(req);
        }

        //声明交换机
        void onDeclareExchange(const muduo::net::TcpConnectionPtr& conn, const declareExchangeRequestPtr& req, muduo::Timestamp)
        {
            //先获取Connection对象
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr)
            {
                DLOG("声明交换机失败，没有找到对应的Connection对象");
                conn->shutdown();
                return;
            }
            //获取信道
            Channel::ptr cp = mconn->getChannel(req->cid());
            if (mconn.get() == nullptr)
            {
                DLOG("声明交换机时，没有找到对应的信道");
                conn->shutdown();
                return;
            }
            DLOG("声明交换机成功");
            return cp->declareExchange(req);
        }

        //删除交换机
        void onDeleteExchange(const muduo::net::TcpConnectionPtr& conn, const deleteExchangeRequestPtr& req, muduo::Timestamp)
        {
            //先获取Connection对象
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr)
            {
                DLOG("删除交换机失败，没有找到对应的Connection对象");
                conn->shutdown();
                return;
            }
            //获取信道
            Channel::ptr cp = mconn->getChannel(req->cid());
            if (mconn.get() == nullptr)
            {
                DLOG("删除交换机时，没有找到对应的信道");
                conn->shutdown();
                return;
            }
            DLOG("删除交换机成功");
            return cp->deleteExchange(req);
        }

        //声明队列
        void onDeclareQueue(const muduo::net::TcpConnectionPtr& conn, const declareQueueRequestPtr& req, muduo::Timestamp)
        {
            //先获取Connection对象
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr)
            {
                DLOG("声明队列失败，没有找到对应的Connection对象");
                conn->shutdown();
                return;
            }
            //获取信道
            Channel::ptr cp = mconn->getChannel(req->cid());
            if (mconn.get() == nullptr)
            {
                DLOG("声明队列时，没有找到对应的信道");
                conn->shutdown();
                return;
            }
            DLOG("声明队列成功");
            return cp->declareQueue(req);
        }
        //删除队列
        void onDeleteQueue(const muduo::net::TcpConnectionPtr& conn, const deleteQueueRequestPtr& req, muduo::Timestamp)
        {
            //先获取Connection对象
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr)
            {
                DLOG("删除队列失败，没有找到对应的Connection对象");
                conn->shutdown();
                return;
            }
            //获取信道
            Channel::ptr cp = mconn->getChannel(req->cid());
            if (mconn.get() == nullptr)
            {
                DLOG("删除队列时，没有找到对应的信道");
                conn->shutdown();
                return;
            }
            DLOG("删除队列成功");
            return cp->deleteQueue(req);
        }

        //队列绑定
        void onQueueBind(const muduo::net::TcpConnectionPtr& conn, const queueBindRequestPtr& req, muduo::Timestamp)
        {
            //先获取Connection对象
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr)
            {
                DLOG("绑定队列失败，没有找到对应的Connection对象");
                conn->shutdown();
                return;
            }
            //获取信道
            Channel::ptr cp = mconn->getChannel(req->cid());
            if (mconn.get() == nullptr)
            {
                DLOG("绑定队列时，没有找到对应的信道");
                conn->shutdown();
                return;
            }
            return cp->queueBind(req);
        }

        //队列解绑
        void onQueueUnBind(const muduo::net::TcpConnectionPtr& conn, const queueUnBindRequestPtr& req, muduo::Timestamp)
        {
            //先获取Connection对象
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr)
            {
                DLOG("队列解绑失败，没有找到对应的Connection对象");
                conn->shutdown();
                return;
            }
            //获取信道
            Channel::ptr cp = mconn->getChannel(req->cid());
            if (mconn.get() == nullptr)
            {
                DLOG("队列解绑时，没有找到对应的信道");
                conn->shutdown();
                return;
            }
            DLOG("队列解绑定成功");
            return cp->queueUnBind(req);
        }

        //消息发布
        void onBasicPublish(const muduo::net::TcpConnectionPtr& conn, const basicPublishRequestPtr& req, muduo::Timestamp)
        {
            //先获取Connection对象
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr)
            {
                DLOG("消息发布失败，没有找到对应的Connection对象");
                conn->shutdown();
                return;
            }
            //获取信道
            Channel::ptr cp = mconn->getChannel(req->cid());
            if (mconn.get() == nullptr)
            {
                DLOG("消息发布时，没有找到对应的信道");
                conn->shutdown();
                return;
            }
            DLOG("消息发布成功");
            return cp->basicPublish(req);
        }

        //消息确认
        void onBasicAck(const muduo::net::TcpConnectionPtr& conn, const basicAckRequestPtr& req, muduo::Timestamp)
        {
            //先获取Connection对象
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr)
            {
                DLOG("消息确认失败，没有找到对应的Connection对象");
                conn->shutdown();
                return;
            }
            //获取信道
            Channel::ptr cp = mconn->getChannel(req->cid());
            if (mconn.get() == nullptr)
            {
                DLOG("消息确认时，没有找到对应的信道");
                conn->shutdown();
                return;
            }
            DLOG("消息确认成功");
            return cp->basicAck(req);
        }

        //队列消息订阅
        void onBasicConsumer(const muduo::net::TcpConnectionPtr& conn, const basicConsumeRequestPtr& req, muduo::Timestamp)
        {
            //先获取Connection对象
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr)
            {
                DLOG("队列订阅消息失败，没有找到对应的Connection对象");
                conn->shutdown();
                return;
            }
            //获取信道
            Channel::ptr cp = mconn->getChannel(req->cid());
            if (mconn.get() == nullptr)
            {
                DLOG("队列订阅消息时，没有找到对应的信道");
                conn->shutdown();
                return;
            }
            DLOG("队列订阅队列成功");
            return cp->basicConsumer(req);
        }

        //队列消息取消订阅
        void onBasicCancel(const muduo::net::TcpConnectionPtr& conn, const basicCancelRequestPtr& req, muduo::Timestamp)
        {
            //先获取Connection对象
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if (mconn.get() == nullptr)
            {
                DLOG("队列消息取消订阅失败，没有找到对应的Connection对象");
                conn->shutdown();
                return;
            }
            //获取信道
            Channel::ptr cp = mconn->getChannel(req->cid());
            if (mconn.get() == nullptr)
            {
                DLOG("队列消息取消订阅时，没有找到对应的信道");
                conn->shutdown();
                return;
            }
            DLOG("取消订阅成功");
            return cp->basicCancel(req);
        }

        void onUnknownMessage(const muduo::net::TcpConnectionPtr&, const MessagePtr& message, muduo::Timestamp)
        {
            std::cout << "onUnknownMessage: " << message->GetTypeName() << std::endl;
        }

        void onConnection(const muduo::net::TcpConnectionPtr& conn)
        {
            if (conn->connected())
            {
                DLOG("连接建立成功");
                _connection_manager->newConnection(_virtual_host, _consumer_manager,
                    _codec, conn, _threadpool);
            }
            else 
            {
                DLOG("连接断开");
                _connection_manager->delConnection(conn);
            }
        }
        
    private:
        muduo::net::EventLoop _baseloop;
        muduo::net::TcpServer _server;//服务器对象
        ProtobufDispatcher _dispatcher;//请求分发器对象--要向其中注册请求处理函数
        ProtobufCodecPtr _codec;//protobuf协议处理器--针对收到的请求数据进行protobuf协议处理
        VirtualHost::ptr _virtual_host;
        ConsumerManager::ptr _consumer_manager;
        ConnectionManager::ptr _connection_manager;
        threadpool::ptr _threadpool;
    };
}